package com.bravo.advanced.fluent.stream2;

import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

public class SimpleStream1<T> {
    private final List<T> source;
    // 通过 Stage 把 filter、map 等中间操作统一起来
    private final List<Stage> stages = new ArrayList<>();

    private SimpleStream1(List<T> source) {
        this.source = source;
    }

    public static <T> SimpleStream1<T> of(List<T> source) {
        return new SimpleStream1<>(source);
    }

    public SimpleStream1<T> filter(Predicate<? super T> predicate) {
        stages.add(new FilterStage<>(predicate));
        return this;
    }

    @SuppressWarnings("unchecked")
    public <R> SimpleStream1<R> map(Function<? super T, ? extends R> mapper) {
        stages.add(new MapStage<>(mapper));
        return (SimpleStream1<R>) this;
    }

    @SuppressWarnings("unchecked")
    public void forEach(Consumer<? super T> action) {
        // 遍历元素
        for (T element : source) {
            // 先执行每一个中间操作stage：filter/map...
            Object result = process(element);
            if (result == null) {
                continue;
            }
            // 再执行终端操作：println
            action.accept((T) result);
        }
    }

    private Object process(Object element) {
        Object result = element;
        for (Stage stage : stages) {
            result = stage.accept(result);
            if (result == null) {
                return null;
            }
        }
        return result;
    }

    // ------ 以下是Stage：包装了中间操作，对外统一展示为Stage ------

    private interface Stage {
        Object accept(Object item);
    }

    private static class FilterStage<T> implements Stage {
        private final Predicate<? super T> predicate;

        FilterStage(Predicate<? super T> predicate) {
            this.predicate = predicate;
        }

        @Override
        @SuppressWarnings("unchecked")
        public Object accept(Object item) {
            System.out.println(item + "_filter");
            if (predicate.test((T) item)) {
                return item;
            } else {
                return null;
            }
        }
    }

    private static class MapStage<T, R> implements Stage {
        private final Function<? super T, ? extends R> mapper;

        MapStage(Function<? super T, ? extends R> mapper) {
            this.mapper = mapper;
        }

        @Override
        @SuppressWarnings("unchecked")
        public Object accept(Object item) {
            System.out.println(item + "_map");
            return mapper.apply((T) item);
        }
    }
}